package com.yanxu;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-05-19-2025
 * @Description: 这里用一句话描述这个类的作用
 */
public class StreamWordCount {

    /*
    * 前期准备：
    *   1: 准备一台 192.168.150.171 centos7 服务器。
    *   2: 安装工具，yum install nmap-ncat -y
    *   3: 监听端口， nc -lk 6666
    * */

    /*
     *  使用 flink 进行流式处理
     * */
    public static void main(String[] args) throws Exception {
        // 创建 Flink 执行环境
        //ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 从socket 中读取文本数据，作为数据源。(此时的数据是一个 流数据)
        DataStreamSource<String> lineDSS = env.socketTextStream("192.168.1.30", 6666);

        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne =
            lineDSS.flatMap(new FlatMapFunction<String, String>() {
                        @Override
                        public void flatMap(String line, Collector<String> words) throws Exception {
                            /**
                             * Hello
                             * World
                             * Thank
                             * you
                             * very
                             * much
                             * Hello
                             * Lily
                             */
                            // 将按空格分割的单词收集到words中
                            String[] wordArray = line.split(" ");
                            for (String word : wordArray) {
                                words.collect(word);
                            }
                        }
                    })
                    // 设置并行度
                    //.setParallelism(3)
                    // 指定flatMap输出的数据类型为String
                    .returns(Types.STRING)
                    /*
                     * 将每个单词映射为一个Tuple2类型，格式为（word, 1L），表示每个单词出现一次
                     * Hello  1
                     * World  1
                     * Thank  1
                     * you   1
                     * very  1
                     * much  1
                     * Hello  1
                     * Lily  1
                    * */
                    .map(word -> Tuple2.of(word, 1L))
                    // 指定map操作输出的数据类型为Tuple2<String, Long>
                    .returns(Types.TUPLE(Types.STRING, Types.LONG));


        // 对数据流进行分组，按照Tuple2的第一个元素（即单词）进行分组
        // 注意：流式数据进行分组，使用的是 keyBy 进行分组。 （ 批处理进行分组，使用的是 groupBy 方法）
        KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOne
                .keyBy(t -> t.f0);

        /**
         * Hello 1
         * World 1
         * Thank 1
         * you 1
         * very 1
         * much 1
         * Hello 2
         * Lily 1
         */

        // 在每个分组内对第二个元素（计数值）进行求和，得到每个单词的总出现次数
        SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
                .sum(1); // 参数1表示对元组的第二个字段求和

        // 打印处理后的结果
        result.print(Thread.currentThread().getName()+" >>>");

        /*
            main >>>:9> (world,1)
            main >>>:5> (hello,1)
            main >>>:9> (zhangsan,1)
            main >>>:15> (lisi,1)
            main >>>:10> (wangwu,1)
            main >>>:5> (zhaoliu,1)
            main >>>:3> (xiaoming,1)
            main >>>:5> (hello,6)
        * 全部都是 main 线程，但是又有并行度，但是这个并行度又不是传统意义上的多线程
        * */

        // 启动Flink流处理任务
        env.execute();

    }
}